﻿///////////////////////////////////////////////////////////////////////////////
//
//  This source is part of the  "Managed Task Framework.NET Runtime"  (MTF.NET Runtime).
//
//
//  Copyright (c) 2011, Multicore Consulting Limited (UK)
//  All rights reserved.
//  http://www.multicoreconsulting.co.uk
//
//  Redistribution and use in source and binary forms, with or without
//  modification, are permitted provided that the following conditions
//  are met:
//
//  -  Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//  -  Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//  -  Neither the name of Multicore Consulting Limited nor the names of
//     its contributors may be used to endorse or promote products
//     derived from this software without specific prior written
//     permission.
//
//  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
//  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
//  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
//  FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
//  COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
//  INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
//  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
//  LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
//  LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
//  ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
//  POSSIBILITY OF SUCH DAMAGE.
//
///////////////////////////////////////////////////////////////////////////////

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

namespace MTFOrg.MTF.MTFRuntime
{
	/// <summary>
	/// Task manager implemented using a thread pool
	/// </summary>
	public class ThreadPoolTaskManager : TaskManager
	{
        /// <summary>
        /// Class representing a work item for the work queue.
        /// </summary>
        protected class WorkItem
        {
            /// <summary>
            /// Call-back to be invoked when the work item is runnable.
            /// </summary>
            public WaitCallback waitCallback;

            /// <summary>
            /// State associated with the work item.
            /// </summary>
            public object state;

            /// <summary>
            /// Constructor.
            /// </summary>
            /// <param name="waitCallback">Call-back invoked when runnable</param>
            /// <param name="state">State associated with call-back</param>
            public WorkItem(WaitCallback waitCallback, object state)
            {
                this.waitCallback = waitCallback;
                this.state = state;
            }
        }

        private class ThreadWorker
        {
            /// <summary>
            /// Number of priorities.
            /// </summary>
            private const int NUM_PRIORITIES = 10;

            /// <summary>
            /// Reference back to the task manager.
            /// </summary>
            private ThreadPoolTaskManager m_TaskMan;

            /// <summary>
            /// Threads owned by the pool.
            /// </summary>
            private Thread m_Thread;

            /// <summary>
            /// Whether the application is still running.
            /// </summary>
            private bool m_IsAlive;

            /// <summary>
            /// Queue of items waiting to run.
            /// </summary>
            private PriorityQueue<WorkItem> m_WorkQueue;

            /// <summary>
            /// Lock protecting the work queue.
            /// </summary>
            private SpinLock m_WorkLock;

            /// <summary>
            /// Semaphore that workers wait on when there is insufficient work.
            /// </summary>
            private Semaphore m_WorkSem;

            /// <summary>
            /// Number of workers that are blocked.
            /// </summary>
            private int m_HeavyWaiters;

            /// <summary>
            /// Number of work items in the queue.
            /// </summary>
            private int m_NumWorkItems;

            /// <summary>
            /// Index of this thread worker.
            /// </summary>
            private int m_ThreadIndex;

            /// <summary>
            /// Flag indicating whether this is a GUI thread.
            /// </summary>
            private bool m_IsGui;

            /// <summary>
            /// Gets the number of work items in the queue.
            /// </summary>
            public int NumWorkItems
            {
                get { return m_NumWorkItems; }
            }

            /// <summary>
            /// Constructor.
            /// </summary>
            public ThreadWorker(ThreadPoolTaskManager taskMan, int threadIndex, ThreadPriority threadPriority, bool isGui)
            {
                m_ThreadIndex = threadIndex;
                m_TaskMan = taskMan;
                m_IsAlive = true;
                m_HeavyWaiters = 0;
                m_NumWorkItems = 0;
                m_WorkQueue = new PriorityQueue<WorkItem>(NUM_PRIORITIES);
                m_WorkLock = new SpinLock();
                m_WorkSem = new Semaphore(0, int.MaxValue);
                m_IsGui = isGui;
                m_Thread = new Thread(threadWorker);
                m_Thread.Priority = threadPriority;
                m_Thread.Start();
            }

            /// <summary>
            /// Gets work from the queue and runs it.
            /// </summary>
            protected void threadWorker()
            {
                // Set up the thread index in thread-local-storage.
                TaskManager.CurrentThreadIndex = m_ThreadIndex;

                while (m_IsAlive)
                {
                    m_WorkLock.Lock();
                    if (m_WorkQueue.Count > 0)
                    {
                        // Grab next item from the queue and run it
                        Interlocked.Decrement(ref m_NumWorkItems);
                        WorkItem workItem = m_WorkQueue.Dequeue();
                        m_WorkLock.Unlock();
                        workItem.waitCallback.Invoke(workItem.state);
                    }
                    else
                    {
                        // Try stealing an item from another queue
                        WorkItem stolenWorkItem = m_TaskMan.StealWorkItem(m_IsGui);
                        if (stolenWorkItem != null)
                        {
                            // Successful steal
                            m_WorkLock.Unlock();
                            stolenWorkItem.waitCallback.Invoke(stolenWorkItem.state);
                        }
                        else
                        {
                            // Steal failed - stall thread and wait for more work
                            m_HeavyWaiters++;
                            m_WorkLock.Unlock();
                            m_WorkSem.WaitOne();
                        }
                    }
                }

                m_TaskMan.NotifyThreadExit();
            }

            /// <summary>
            /// Steals a work item from this thread worker if it has one.
            /// </summary>
            /// <returns>Work item if available, otherwise null</returns>
            public WorkItem StealWorkItem()
            {
                WorkItem workItem = null;

                m_WorkLock.Lock();
                if (m_WorkQueue.Count > 0)
                {
                    Interlocked.Decrement(ref m_NumWorkItems);
                    workItem = m_WorkQueue.Dequeue();
                }
                m_WorkLock.Unlock();

                return workItem;
            }

            /// <summary>
            /// Adds new work to the queue.
            /// </summary>
            /// <param name="waitCallback"></param>
            /// <param name="state"></param>
            public void queueUserWorkItem(WaitCallback waitCallback, object state)
            {
                queueUserWorkItem(waitCallback, state, NUM_PRIORITIES - 1);
            }

            /// <summary>
            /// Adds new work to the queue.
            /// </summary>
            /// <param name="waitCallback"></param>
            /// <param name="state"></param>
            /// <param name="priority">Execution priority of the work item</param>
            public void queueUserWorkItem(WaitCallback waitCallback, object state, int priority)
            {
                m_WorkLock.Lock();
                m_WorkQueue.Enqueue(new WorkItem(waitCallback, state), priority);
                Interlocked.Increment(ref m_NumWorkItems);
                if (m_HeavyWaiters > 0)
                {
                    m_WorkSem.Release();
                    m_HeavyWaiters--;
                }
                m_WorkLock.Unlock();
            }

            /// <summary>
            /// Terminates execution of the worker thread.
            /// <param name="force">forces workers to abort immediately</param>
            /// </summary>
            public void Terminate(bool force = false)
            {
                m_IsAlive = false;

                if (force)
                    m_Thread.Abort();
            }
        }

        /// <summary>
        /// Worker threads.
        /// </summary>
        private ThreadWorker[] m_ThreadWorkers;

        /// <summary>
        /// Gui worker threads.
        /// </summary>
        private ThreadWorker[] m_GuiThreadWorkers;

        /// <summary>
        /// Number of threads in the pool.
        /// </summary>
        private int m_NumThreads;

        /// <summary>
        /// Number of threads that have exited.
        /// </summary>
        private int m_NumExitedThreads;

        /// <summary>
        /// Determines if the threads are initialized.
        /// </summary>
        private bool m_IsInitialized;

        /// <summary>
        /// Callback function for asynchronous message sends.
        /// </summary>
        private WaitCallback m_QueueMsgCallback;
        
        /// <summary>
        /// Constructor.
        /// </summary>
        public ThreadPoolTaskManager(ThreadPriority guiThreadPriority)
        {
            m_QueueMsgCallback = new WaitCallback(AsyncActionMsg);
            m_IsInitialized = false;
            m_NumExitedThreads = 0;
            m_NumThreads = Environment.ProcessorCount;
            m_ThreadWorkers = new ThreadWorker[m_NumThreads];
            for (int i = 0; i < m_NumThreads; i++)
            {
                m_ThreadWorkers[i] = new ThreadWorker(this, i, ThreadPriority.BelowNormal, false);
            }
            m_GuiThreadWorkers = new ThreadWorker[m_NumThreads];
            for (int i = 0; i < m_NumThreads; i++)
            {
                m_GuiThreadWorkers[i] = new ThreadWorker(this, i, guiThreadPriority, true);
            }
            m_IsInitialized = true;
        }

        /// <summary>
        /// Attempts to find a queue with excess work items and steals an item from it.
        /// </summary>
        /// <param name="isGui">Whether to steal from the GUI thread pool</param>
        /// <returns>Stolen work item, or null if none could be stolen</returns>
        protected WorkItem StealWorkItem(bool isGui)
        {
            WorkItem stolenWorkItem = null;

            if (m_IsInitialized)
            {
                int busiestQueueIndex = GetBusiestQueueIndex(isGui);
                ThreadWorker[] threadWorkers = isGui ? m_GuiThreadWorkers : m_ThreadWorkers;
                if (threadWorkers[busiestQueueIndex].NumWorkItems > 0)
                {
                    stolenWorkItem = threadWorkers[busiestQueueIndex].StealWorkItem();
                }
            }

            return stolenWorkItem;
        }

        /// <summary>
        /// Gets the index of the queue with the most items.
        /// </summary>
        /// <returns>Index of queue with most items</returns>
        protected int GetBusiestQueueIndex(bool isGui)
        {
            int busiestQueueIndex = 0;
            int maxQueueItems = 0;

            ThreadWorker[] threadWorkers = isGui ? m_GuiThreadWorkers : m_ThreadWorkers;

            for (int i = 0; i < m_NumThreads; i++)
            {
                int numWorkItems = threadWorkers[i].NumWorkItems;
                if (numWorkItems > maxQueueItems)
                {
                    busiestQueueIndex = i;
                    maxQueueItems = numWorkItems;
                }
            }

            return busiestQueueIndex;
        }

        /// <summary>
        /// Gets the index of the queue with the fewest items.
        /// </summary>
        /// <returns>Index of queue with fewest items</returns>
        protected int GetEmptiestQueueIndex()
        {
            int emptiestQueueIndex = 0;
            int minQueueItems = int.MaxValue;

            for (int i = 0; i < m_NumThreads; i++)
            {
                int numWorkItems = m_ThreadWorkers[i].NumWorkItems;
                if (numWorkItems < minQueueItems)
                {
                    emptiestQueueIndex = i;
                    minQueueItems = numWorkItems;
                    if (minQueueItems == 0) break;
                }
            }

            return emptiestQueueIndex;
        }

        /// <summary>
        /// Gets the index of the queue with the fewest items.
        /// </summary>
        /// <returns>Index of queue with fewest items</returns>
        protected int GetEmptiestGuiQueueIndex()
        {
            int emptiestQueueIndex = 0;
            int minQueueItems = int.MaxValue;

            for (int i = 0; i < m_NumThreads; i++)
            {
                int numWorkItems = m_GuiThreadWorkers[i].NumWorkItems;
                if (numWorkItems < minQueueItems)
                {
                    emptiestQueueIndex = i;
                    minQueueItems = numWorkItems;
                    if (minQueueItems == 0) break;
                }
            }

            return emptiestQueueIndex;
        }

        /// <summary>
        /// Disposes of the task manager and its resources.
        /// <param name="force">forces workers to abort immediately</param>
        /// </summary>
        public override void Dispose(bool force = false)
        {
            // Notify pre-shutdown
            if (OnPreShutdown != null)
                OnPreShutdown();

            // Request all worker threads to terminate
            for (int i = 0; i < m_NumThreads; i++)
            {
                m_ThreadWorkers[i].Terminate(force);
                m_GuiThreadWorkers[i].Terminate(force);
            }

            if (!force)
            {
                // Insert dummy tasks to prevent threads remaining blocked on their work semaphore
                for (int i = 0; i < m_NumThreads; i++)
                {
                    m_ThreadWorkers[i].queueUserWorkItem(new WaitCallback(delegate(object cxnFiredArgs) { }), null, 0);
                    m_GuiThreadWorkers[i].queueUserWorkItem(new WaitCallback(delegate(object cxnFiredArgs) { }), null, 0);
                }
            }
        }

        /// <summary>
        /// Delegate for a thread exited e
        /// </summary>
        public delegate void PostShutdownDelegate();

        /// <summary>
        /// Holds the shutdown completed call-back.
        /// </summary>
        public static event PostShutdownDelegate OnPostShutdown;

        /// <summary>
        /// Delegate for a thread exited e
        /// </summary>
        public delegate void PreShutdownDelegate();

        /// <summary>
        /// Holds the shutdown completed call-back.
        /// </summary>
        public static event PreShutdownDelegate OnPreShutdown;

        /// <summary>
        /// Called when a worker thread is exited.
        /// </summary>
        public void NotifyThreadExit()
        {
            // Increment exited threads count
            Interlocked.Increment(ref m_NumExitedThreads);

            // If shutdown is complete and a handler is provided then invoke it
            if (IsShutdown() && OnPostShutdown != null)
            {
                OnPostShutdown();
            }
        }

        /// <summary>
        /// Determines if shutdown of the worker threads has completed.
        /// </summary>
        /// <returns>true if shutdown complete, otherwise false</returns>
        public override bool IsShutdown()
        {
            return m_NumExitedThreads == m_NumThreads * 2;
        }

        /// <summary>
        /// Adds new work to the queue for the worker thread with the lightest load.
        /// </summary>
        /// <param name="waitCallback">Call-back executed by the worker thread</param>
        /// <param name="state">State to pass to the call-back function</param>
        protected void QueueUserWorkItem(WaitCallback waitCallback, object state)
        {
            int nextThreadIndex = GetEmptiestQueueIndex();

            m_ThreadWorkers[nextThreadIndex].queueUserWorkItem(waitCallback, state);
        }

        /// <summary>
        /// Adds new work to the queue for the worker thread with the lightest load.
        /// </summary>
        /// <param name="waitCallback">Call-back executed by the worker thread</param>
        /// <param name="state">State to pass to the call-back function</param>
        /// <param name="priority">Execution priority of the work item</param>
        public override void QueueUserWorkItem(WaitCallback waitCallback, object state, int priority)
        {
            int nextThreadIndex = GetEmptiestQueueIndex();

            m_ThreadWorkers[nextThreadIndex].queueUserWorkItem(waitCallback, state, priority);
        }

        /// <summary>
        /// Adds new work to the queue for a specific worker thread.
        /// </summary>
        /// <param name="waitCallback">Call-back executed by the worker thread</param>
        /// <param name="state">State to pass to the call-back function</param>
        /// <param name="priority">Execution priority of the work item</param>
        /// <param name="threadIndex">Index of thread to run task on</param>
        /// <param name="isGui">Whether the item is to run in the GUI worker pool</param>
        public override void QueueUserWorkItem(WaitCallback waitCallback, object state, int priority, int threadIndex, bool isGui)
        {
            if (!isGui)
            {
                if (threadIndex == -1) threadIndex = GetEmptiestQueueIndex();
                m_ThreadWorkers[threadIndex].queueUserWorkItem(waitCallback, state, priority);
            }
            else
            {
                int nextThreadIndex = GetEmptiestGuiQueueIndex();
                m_GuiThreadWorkers[nextThreadIndex].queueUserWorkItem(waitCallback, state, priority);
            }
        }

        /// <summary>
        /// Synchronous event send
        /// </summary>
        /// <param name="newMesage">Message to send</param>
        /// <param name="connection">Connection to send event on</param>
        public override void SendMessage(IMessage newEvent, IConnection connection)
        {
            LocatedConnection locatedConnection = connection as LocatedConnection;

#if DEBUG
            if (locatedConnection == null)
                throw new System.Exception("Attempt to send on non-located connection.");
#endif

            ITask targetTask = locatedConnection.ReceiverConnectionPoint.ParentTask;

            // Send event
            targetTask.ProcessEvent(new ConnectionFiredEventArgs(connection, newEvent));
        }

        private void AsyncActionMsg(object cxnFiredArgs)
        {
            ConnectionFiredEventArgs cfeArgs = (ConnectionFiredEventArgs)cxnFiredArgs;

            LocatedConnection locatedConnection = cfeArgs.Connection as LocatedConnection;
#if DEBUG
            if (locatedConnection == null)
                throw new System.Exception("Attempt to send on non-located connection.");
#endif

            ITask targetTask = locatedConnection.ReceiverConnectionPoint.ParentTask;

            targetTask.ProcessEvent(cfeArgs);
        }

        /// <summary>
        /// Asynchronous event send
        /// </summary>
        /// <param name="newMessage">Message to send</param>
        /// <param name="connection">Connection to send event on</param>
        public override void QueueMessage(IMessage newMessage, IConnection connection)
        {
            ConnectionFiredEventArgs args = new ConnectionFiredEventArgs(connection, newMessage);
            QueueUserWorkItem(m_QueueMsgCallback, args, 0);
        }

        /// <summary>
        /// Asynchronous event send
        /// </summary>
        /// <param name="newMessage">Message to send</param>
        /// <param name="connection">Connection to send event on</param>
        /// <param name="threadIndex">Index of preferred thread for execution</param>
        public override void QueueMessage(IMessage newMessage, IConnection connection, int threadIndex)
        {
            ConnectionFiredEventArgs args = new ConnectionFiredEventArgs(connection, newMessage);
            QueueUserWorkItem(m_QueueMsgCallback, args, 0, threadIndex, false);
        }
    }
}
